/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * htap_modify.cpp
 * Interface for main HTAP hooks.
 *
 * IDENTIFICATION
 * src/gausskernel/storage/htap/htap_modify.cpp
 *
 * -------------------------------------------------------------------------
 */

#include "libintl.h"
#include "postgres.h"
#include "htap_modify.h"
#include "htap_infra.h"
#include "htap_helpers.h"
#include "htap_logger.h"
#include "htap_internal.h"
#include "htap_statistics.h"
#include "cycles.h"
#include "utils/elog.h"
#include "pgxc/pgxc.h"

HTAP_DECLARE_LOGGER(HTAPHooks, HTAP)

// refactor common code and use lambda expressions
template <typename HookFunc, typename StatFunc>
static void HTAPIUDHookImpl(const char *opName, HookFunc hookFunc, StatFunc statFunc)
{
    PG_TRY();
    {
        if (IS_PGXC_DATANODE) {
            htap::HTAPInfra *htap = htap::HTAPInfra::GetInstance();
            if (unlikely(htap == nullptr)) {
                HTAP_LOG_TRACE("HTAPInfra is null, %s hook discarded", opName);
            } else if (unlikely(!htap->GetCfg().m_enableHtap)) {
                HTAP_LOG_DEBUG("HTAP is disabled, %s hook discarded", opName);
            } else {
                // invoke hook
                uint64_t startTime = GetSysClock();
                htap::RC result = hookFunc(htap);
                uint64_t endTime = GetSysClock();

                // update statistics (even if error occurred)
                uint64_t timeMicros = MOT::CpuCyclesLevelTime::CyclesToMicroseconds(endTime - startTime);
                htap::ModifyStatisticsProvider &stats = htap::ModifyStatisticsProvider::GetInstance();
                statFunc(stats, timeMicros, result);

                // print error stack if needed
                if ((result != htap::HTAP_RC_OK) && (result != htap::HTAP_RC_EVENT_DISCARDED)) {
                    htap::PrintErrorStack();
                    HTAP_LOG_ERROR_EX("Failed to execute %s hook: %s", opName, htap::RcToString(result));
                }
            }
        }
    }
    PG_CATCH();
    {
        HTAP_LOG_ERROR("Encountered exception while executing %s: %s (%d)", opName, Geterrmsg(), geterrcode());
        PG_RE_THROW();
    }
    PG_END_TRY();
}

// refactor common code and use lambda expressions
template <typename HookFunc> static void HTAPDDLHookImpl(const char *opName, HookFunc hookFunc)
{
    PG_TRY();
    {
        if (IS_PGXC_DATANODE) {
            htap::HTAPInfra *htap = htap::HTAPInfra::GetInstance();
            if (unlikely(htap == nullptr)) {
                HTAP_LOG_TRACE("HTAPInfra is null, %s hook discarded", opName);
            } else if (unlikely(!htap->GetCfg().m_enableHtap)) {
                HTAP_LOG_DEBUG("HTAP is disabled, %s hook discarded", opName);
            } else {
                // invoke hook
                htap::RC result = hookFunc(htap);
                if (result != htap::HTAP_RC_OK) {
                    htap::PrintErrorStack();
                    HTAP_LOG_ERROR_EX("Failed to execute %s hook: %s", opName, htap::RcToString(result));
                }
            }
        }
    }
    PG_CATCH();
    {
        HTAP_LOG_ERROR("Encountered exception while executing %s: %s (%d)", opName, Geterrmsg(), geterrcode());
        PG_RE_THROW();
    }
    PG_END_TRY();
}

inline bool IsIMCStoredTable(Oid relationId)
{
    MOT::TxnManager *txn = htap::HTAPGetSafeTxn();
    return (txn->GetTableByExternalId(relationId) != nullptr);
}

void HTAPExecInsertRow(Oid relationId, ItemPointer ctid, TupleTableSlot *slot)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        return;
    }

    htap::HTAPInfra *htap = htap::HTAPInfra::GetInstance();
    if (unlikely(htap == nullptr)) {
        HTAP_LOG_TRACE("HTAPInfra not initialized");
        return;
    }

    if (unlikely(!htap->checkBeforeModify())) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAP Insert Row Error", "Failed to execute insert row actions");
        return;
    }

    htap::RC result = htap::HTAPAdaptor::TableInsert(relationId, ctid, slot);
    if (result != htap::HTAP_RC_OK) {
        htap::PrintErrorStack();
        HTAP_LOG_ERROR_EX("Failed to exec insert row: %s", htap::RcToString(result));
    }
}

void HTAPExecDeleteRow(Oid relationId, ItemPointer ctid)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        return;
    }

    htap::HTAPInfra *htap = htap::HTAPInfra::GetInstance();
    if (unlikely(htap == nullptr)) {
        HTAP_LOG_TRACE("HTAPInfra not initialized");
        return;
    }

    if (unlikely(!htap->checkBeforeModify())) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAP Delete Row Error", "Failed to execute delete row actions");
        return;
    }

    htap::RC result = htap::HTAPAdaptor::TableDelete(relationId, ctid, false);
    if (result != htap::HTAP_RC_OK) {
        htap::PrintErrorStack();
        HTAP_LOG_ERROR_EX("Failed to exec delete row: %s", htap::RcToString(result));
    }
}

void HTAPExecUpdateRow(Oid relationId, ItemPointer oldCtid, ItemPointer newCtid, TupleTableSlot *slot)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        return;
    }

    htap::HTAPInfra *htap = htap::HTAPInfra::GetInstance();
    if (unlikely(htap == nullptr)) {
        HTAP_LOG_TRACE("HTAPInfra not initialized");
        return;
    }

    if (unlikely(!htap->checkBeforeModify())) {
        HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "HTAP Update Row Error", "Failed to execute update row actions");
        return;
    }

    htap::RC result;
    if (ItemPointerEquals(oldCtid, newCtid)) {
        result = htap::HTAPAdaptor::TableModify(relationId, oldCtid, slot);
    } else {
        result = htap::HTAPAdaptor::TableDelete(relationId, oldCtid, true);
        if (result == htap::HTAP_RC_OK) {
            result = htap::HTAPAdaptor::TableInsert(relationId, newCtid, slot);
        }
    }
    if (result != htap::HTAP_RC_OK) {
        htap::PrintErrorStack();
        HTAP_LOG_ERROR_EX("Failed to exec update row: %s", htap::RcToString(result));
    }
}

void HTAPCreateTable(Oid parentID, Oid relationId)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        return;
    }

    HTAPDDLHookImpl("create-table",
                    [&](htap::HTAPInfra *htap) { return htap->OnCreateTable(parentID, relationId); });
}

void HTAPDropTable(Oid relationId, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        return;
    }

    HTAPDDLHookImpl("drop-table",
        [&](htap::HTAPInfra *htap) { return htap->OnDropTable(relationId, out_oid_del_info); });
}

void HTAPTruncateTable(Oid relationId, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        return;
    }

    HTAPDDLHookImpl("truncate-table",
        [&](htap::HTAPInfra *htap) { return htap->OnTruncateTable(relationId, out_oid_del_info); });
}

void HTAPAlterTable(Oid relationId, Relation rel, AlterTableStmt *stmt)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        return;
    }

    HTAPDDLHookImpl("alter-table", [&](htap::HTAPInfra *htap) { return htap->OnAlterTable(relationId, rel, stmt); });
}

void HTAPSearchDelta(Oid relationId, uint64_t &rowCount, uint64_t &visibleRowCount, uint64_t &memSize)
{
    if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
        return;
    }

    HTAPDDLHookImpl("query-delta",
        [&](htap::HTAPInfra *htap) { return htap->OnSearchDelta(relationId, rowCount, visibleRowCount, memSize); });
}
